package com.ibm.cps.processors;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;

import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.message.ParsedDataMessageBuilder;
import com.ibm.cps.sql.expression.function.meta.FieldNameParam;
import com.ibm.cps.sql.expression.function.meta.IgnoredInFunctionCall;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.ErrorCode;
import com.ibm.util.exception.CPSException;

/**
 * Created by telekinesis on 6/18/15.
 */
@Info(Name=MessageFields.AVERAGE_TYPE,Type=Info.ProcessorType.Vector)
public class AverageProcessor extends SchemaInheritiveProcessor {
	@FieldNameParam
	private String inputField;
	@FieldNameParam @IgnoredInFunctionCall
	private String outputField;

	public AverageProcessor(String tenantId, String processorId, String source, ITimeSeriesSchema parentSchema) throws CPSException {
		super(tenantId, processorId, source, parentSchema);
	}

	@Override
	public void dispose() {

	}


	@Override
	protected List<NewField> defineFieldsToBeAdded() {
		List<NewField> fields = new ArrayList<>();
		NewField f = new NewField(outputField, "double");
		fields.add(f);
		return fields;
	}

	@Override
	protected Collection<ParsedDataMessage> runProcessorLogic(ITimeSeriesSchema parentSchema,
			ITimeSeriesSchema outputSchema, ParsedDataMessage message) throws CPSException {
		Collection<ParsedDataMessage> messages = new ArrayList<>();
		ParsedDataMessageBuilder outputBuilder = ParsedDataMessageBuilder.createInhertiveBuilder(outputSchema, message, parentSchema);
		int index = parentSchema.getTagId(inputField);
//		Object[] input =  message.getArrayValue(index);
		Double[] input =  message.getArray(index);
		if (input == null || input.length == 0) {
			throw new CPSException(ErrorCode.INVALID_INPUT_ERROR, 
					"input's length was wrong " + Arrays.toString(input));
		}
		double sum = 0;
		for(Object e : input){
			sum += (double)e;
		}
		double avg = sum / input.length;
		outputBuilder.append(outputField, avg);
		messages.add(outputBuilder.toParsedDataMessage());
		return messages;
	}

	@Override
	public void initialize() throws CPSException {
		
	}
}
